// pipe.h : 
// 

#ifndef _PIPE_H_KOG_
#define _PIPE_H_KOG_

#include <queue>
#include <memory>
#include "basictypes.h"
#include "winobject.h"

NAMESPACE_BEGIN(kog);

class pipe
{
public:
	typedef kog::os::osobject winobject;
public:
	enum port_type
	{
		server_port,
		client_port,
	};

	enum
	{
		MAX_META_DATA = 4 * 1024, // max size of a meta data -> 4KB
		MAX_QUEUE_SIZE = 10000
	};

	struct msgpeice_t
	{
		const char* ptr; // content's address
		int len; // length of the data
	};

	typedef msgpeice_t pipemeta;
private:
	struct message_t
	{
		// length of data
		// if message is to send, len == sizeof(message_t) - sizeof(data) + sizeof(usr data)
		// else len == sizeof(usr data)
		int len; 
		int from; // where message from
		int to; // message to where
		int dtype; // data type : mark different application's data
		union 
		{
			char quick[4]; // if message is to send, use this
			char* ptr; // if message is read from pipe, use this
		}data;
	};
public:
	pipe(const std::string& pipe_name, port_type ptype);

	// stop send/recv data and close handle
	virtual ~pipe(void); // close 
private:
	pipe(const pipe&);
	pipe& operator=(const pipe&);
public:
	// is pipe handle is validate?
	bool is_valid() const;
public:
	// connect to the given server
	int connectserver(int timeout = -1);

	// wait client
	int waitclient(int timeout = -1);
public:
	// send data
	int send(int fid, int tid, int dtype, const msgpeice_t* ps, int n);

	// recv data
	int recv(int& fid, int& tid, int& dtype, msgpeice_t& msg, int timeout = -1);

	template<typename _T> int send(int fid, int tid, int dtype, const _T& v)
	{
		msgpeice_t mp = {(const char*)&v, sizeof(v)};
		return send(fid, tid, dtype, &mp, 1);
	}
protected:
	void _send_thread();
	void _recv_thread();
protected:
	void _cleanup();
	int _check(port_type ptype, bool chk_handle);
	void _start_threads();
	void _combine(const msgpeice_t* ps, int n, message_t*& result);
private:
	unsigned long _waitfor(winobject hobject);
private:
	// pipe name
	std::string _pipe_name;
	// server or client ?
	port_type _ptype;

	// pipe handle
	winobject _hpipe;

	// send queue
	std::queue<message_t*> _tosend;
	winobject _send_sem;
	winobject _send_guard;

	// recv buffer
	std::queue<message_t> _recved;
	winobject _recv_sem;
	winobject _recv_guard;

	// is running?
	volatile bool _isruning;

	// send and recv thread handle
	winobject _hsend_thread;
	winobject _hrecv_thread;
};

NAMESPACE_END(kog);

#endif
